package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * 生产者
 */
public class ProducerQuickStart {

    public static void main(String[] args) throws Exception {

        // 1.kafka连接配置信息
        Properties properties = new Properties();
        // kafka连接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.200.130:9092");
        // key和value序列化
        //properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // ack配置 消息确认机制
        /*
        | acks=0           | 生产者在成功写入消息之前不会等待任何来自服务器的响应,消息有丢失的风险，但是速度最快
        | acks=1（默认值）   | 只要集群首领节点收到消息，生产者就会收到一个来自服务器的成功响应
        | acks=all         | 只有当所有参与赋值的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应
         */
        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // 设置重试次数
        /*
        生产者从服务器收到的错误有可能是临时性错误，在这种情况下，retries参数的值决定了生产者可以重发消息的次数，
        如果达到这个次数，生产者会放弃重试返回错误，默认情况下，生产者会在每次重试之间等待100ms
         */
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);

        /*
         消息压缩 默认情况下， 消息发送时不会被压缩
        | snappy | 占用较少的  CPU，  却能提供较好的性能和相当可观的压缩比， 如果看重性能和网络带宽，建议采用
        | lz4    | 占用较少的 CPU， 压缩和解压缩速度较快，压缩比也很客观
        | gzip   | 占用较多的  CPU，但会提供更高的压缩比，网络带宽有限，可以使用这种算法

        使用压缩可以降低网络传输开销和存储开销，而这往往是向 Kafka 发送消息的瓶颈所在。
         */

        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");


        // 2.创建kafka生产对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 3.发送消息
        /**
         * param1:topic
         * param2:消息的key
         * param3:消息的value
         *
         * 不指定key为轮询策略
         * 指定key为按键保存策略
         */
        //ProducerRecord<String, String> producerRecord = new ProducerRecord<>("topic-first", "key-001", "hello,kafka!");
        for (int i = 0; i < 10; i++) {
            
            ProducerRecord<String, String> kvProducerRecord = new ProducerRecord<>("leadnews-topic-input", "hello kafka");
            kafkaProducer.send(kvProducerRecord);
        }
        // 同步发送消息
        /*RecordMetadata recordMetadata = kafkaProducer.send(producerRecord).get();
        System.out.println(recordMetadata.offset());*/

        // 异步消息发送（优选）
       /* kafkaProducer.send(producerRecord, (recordMetadata, e) -> {
            if (e != null) {
                System.out.println("记录异常信息到日志表中");
            }
            System.out.println(recordMetadata.offset());
        });*/

        // 4.关闭消息通道 必须关闭 否则消息发送不成功
        kafkaProducer.close();
    }
}